module dactor.eventlet;

private import core.sys.posix.sys.time;
private import core.sys.posix.sys.socket;
private import core.sys.posix.pthread;
private import core.sys.posix.unistd;
private import core.thread;
private import core.stdc.errno;
private import core.stdc.string;
private import libevent.libevent;
private import dactor.coroutine;
private import dactor.scheduler;
private import dactor.log;
private import linux.aio;

enum {
    SIGEV_THREAD_ID = 4
}

class Event
{
    public enum {
        SUCCESS = 0,
        READ = 1,
        WRITE = 2,
        TIMEOUT = 4,
        SIGNAL = 8,
        PERSIST = 16,
        ERROR = 32
    }

    protected event ev;
    private Coroutine waiter;

    private int fd;
    private short mask;
    private int timeout;

    package int result;

    public this(int fd, int mask, int timeout) {
        this.fd = fd;

        if (mask & READ)    this.mask |= EV_READ;
        if (mask & WRITE)   this.mask |= EV_WRITE;
        if (mask & TIMEOUT) this.mask |= EV_TIMEOUT;
        if (mask & SIGNAL)  this.mask |= EV_SIGNAL;
        if (mask & PERSIST) this.mask |= EV_PERSIST;

        this.timeout = timeout;
        if (timeout >= 0) {
            this.mask |= EV_TIMEOUT;
        }
        event_set(&ev, this.fd, this.mask, &Event.__proc, cast(void*)this);
    }

    public ~this() {
        EventManager.instance.removeEvent(this);
    }

    public int wait() {
        if (!event_pending(&ev, 0, null)) {
            EventManager.instance.addEvent(this);
        }
        result = Event.SUCCESS;
        waiter = Scheduler.instance.getCurrent();
        waiter.status = Coroutine.Status.WAITING;
        Fiber.yield();

        return result;
    }

    protected void onEvent(int events) {
        result = events;
        Scheduler.instance.schedule(waiter);
    }

    private extern(C) static void __proc(int fd, short events, void* arg) {
        Event event = cast(Event)arg;
        // assert(fd == event.fd);
        assert(!(events & EV_PERSIST));

        int events_ = 0;
        if (events & EV_READ) events_ |= Event.READ;
        if (events & EV_WRITE) events_ |= Event.WRITE;
        if (events & EV_TIMEOUT) events_ |= Event.TIMEOUT;
        if (events & EV_SIGNAL) events_ |= Event.SIGNAL;

        event.onEvent(events_);

        // process event first. If NOT, libevent will dead loop
        Coroutine.yield();
    }
}

class TimerEvent : public Event
{
    public this(int timeout) {
        super(-1, Event.TIMEOUT, timeout);
    }
}

class SignalEvent : public Event
{
    public this(int sig, int mask=0, int timeout=-1) {
        super(sig, mask | Event.SIGNAL, timeout);
    }
}

class AioEvent(int SIGNO) : public Event
{
    char[]   buffer;
    size_t   offset;
    aiocb    cb;
    ssize_t  aioRet;
    int      aioError;

    public this(int fd, char[] buffer, int mask, int timeout=-1) {
        this.buffer = buffer;
        super(fd, mask, timeout);
        event_set(&ev, -1, EV_TIMEOUT, &AioEvent.__proc, cast(void*)this);
    }

    public override int wait() {

        waiter = Scheduler.instance.getCurrent();

        memset(&cb, 0, cb.sizeof);
        cb.aio_fildes = fd;
        cb.aio_buf    = buffer.ptr;
        cb.aio_nbytes = buffer.length;
        cb.v64.aio_offset = offset;
        cb.aio_sigevent.sigev_notify = SIGEV_SIGNAL;
        cb.aio_sigevent.sigev_signo  = 35;
        cb.aio_sigevent.sigev_value.sival_ptr = cast(void*)waiter;

        int ret = (mask & EV_READ) ? aio_read(&cb) : aio_write(&cb);
        if (ret != 0) {
            result = Event.ERROR;
            aioError = aio_error(&cb);
            return result;
        }

        waiter.status = Coroutine.Status.WAITING;

        if (timeout >= 0) {
            if (!event_pending(&ev, 0, null)) {
                EventManager.instance.addEvent(this);
            }
            // wait for timeout
            super.wait();
            EventManager.instance.removeEvent(this);
        }
        else {
            Fiber.yield();
        }

        aioError = aio_error(&cb);
        assert(aioError == 0 || aioError == ECANCELED);

        if (aioError == 0) {
            result = Event.SUCCESS;
            aioRet = aio_return(&cb);
        }
        else if (aioError == ECANCELED) {
            result = Event.TIMEOUT;
        }

        return result;
    }

    override void onEvent(int events) {
        assert(events == Event.TIMEOUT);

        int ret = aio_cancel(fd, &cb);
        if (ret == AIO_CANCELED) {
            Scheduler.instance.schedule(waiter, true);
        }
    }

    //private extern(C) static void __aioEventCb(sigval sv) {
    private extern(C) static void __aioEventCb(int, siginfo_t* info, void* arg) {
        int handle = EventManager.instance.getAdapterHandle();
        sigval sv = info._sifields._rt.si_sigval;
        while(true) {
            int n = .send(handle, &sv.sival_ptr, sv.sival_ptr.sizeof, 0);
            if (n == -1) {
                if (core.stdc.errno.getErrno() == EINTR) {
                    continue;
                }
                char* s = strerror(core.stdc.errno.getErrno());
                logError("send failed: %s", s[0..strlen(s)]);
                return;
            }
            assert(n == sv.sival_ptr.sizeof);
            break;
        }
    }

    public static this() {
        logDebug("install signal action");
        sigaction_t sa;
        memset(&sa, 0, sa.sizeof);
        sa.sa_flags = SA_SIGINFO;
        sa.sa_sigaction = &__aioEventCb;
        sigaction(SIGNO, &sa, null);
    }
}

class SafeScheduleAdapter
{
    private int[2] fd;

    public int getAnotherHandle() const {
        return fd[1];
    }

    public this() {
        fd[0] = fd[1] = -1;
    }

    public int open() {
        if (socketpair(AF_UNIX, SOCK_STREAM, 0, fd) != 0) {
            logDebug("socket pair failed: %s", strerror(core.stdc.errno.getErrno()));
            return -1;
        }
        return 0;
    }

    public void close() {
        .close(fd[0]);
        .close(fd[1]);
        fd[0] = fd[1] = -1;
    }

    public void run() {
        auto Event e = new Event(fd[0], Event.READ | Event.PERSIST, -1);
        EventManager.instance.addEvent(e);

        while (true) {

            int r = e.wait();
            assert(r == Event.READ);

            void* ptr;
            r = recv(fd[0], &ptr, ptr.sizeof, MSG_WAITALL);
            if (r == -1 && core.stdc.errno.getErrno() == EINTR) {
                continue;
            }
            if (r != ptr.sizeof) {
                logDebug("recv fail: %s", strerror(core.stdc.errno.getErrno()));
                break;
            }
            Coroutine c = cast(Coroutine)ptr;
            Scheduler.instance.schedule(c, true);
        }
    }
}

class EventManager
{
    private event_base* evb;
    SafeScheduleAdapter adapter;

    public this() {
        evb = event_init();
        adapter = new SafeScheduleAdapter;
        if (adapter.open() != 0) {
            throw new Exception("open adapter fail");
        }
        spawn(&adapter.run);
    }

    public ~this() {
        // event_free(evb);
    }

    public int getAdapterHandle() const {
        return adapter.getAnotherHandle();
    }

    public void addEvent(Event e) {
        timeval* ptv = null;
        timeval timeout = {e.timeout / 1000, (e.timeout % 1000) * 1000};
        if (e.timeout >= 0 || e.mask & EV_TIMEOUT) {
            ptv = &timeout;
        }
        event_base_set(evb, &e.ev);
        if (event_add(&e.ev, ptv) != 0) {
            // LOG_ERROR("register event fail");
            return;
        }
    }

    public void removeEvent(Event e) {
        event_del(&e.ev);
    }

    public void run() {
        while (true) {
            while (Scheduler.instance.hasTasks()) {
                event_base_loop(evb, EVLOOP_NONBLOCK);
                Coroutine.yield();
            }
            event_base_loop(evb, EVLOOP_ONCE);
        }
    }

    /////////////////////////////////////////
    private static EventManager _instance;

    public static this() {
        _instance = new EventManager;
        spawn(&EventManager.instance.run);
    }
    
    public static EventManager instance() {
        return _instance;
    }
}

